package com.intct.func;

import com.alibaba.fastjson.JSONObject;
import com.intct.hbase.bean.LocusBean;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author gufg
 * @since 2025-07-04 10:36
 */
public class DwdLocusProcessFunc extends ProcessFunction<String, LocusBean> {
    @Override
    public void processElement(String jsonStr, ProcessFunction<String, LocusBean>.Context ctx,
                               Collector<LocusBean> out) throws Exception {
        // ETL
                /*
                {
                  "op": "r",
                  "name": "locus",
                  "after": {
                    "driver_id": "11",
                    "lng": "104.04875",
                    "time": "1475976904",
                    "order_id": "17592719043682",
                    "lat": "30.70046"
                  },
                  "ts_ms": 1751439771322
                }
                 */
        JSONObject jsonObj = JSONObject.parseObject(jsonStr);
        JSONObject afterJsonObj = jsonObj.getJSONObject("after");

        String driverId = afterJsonObj.getString("driver_id");

        // 声明一个建造者
        LocusBean.LocusBeanBuilder builder = LocusBean.builder();


        // 脱敏
        String orderId = afterJsonObj.getString("order_id");
        String orderIdDesensitization = orderId.substring(0, 3) + "********" + orderId.substring(11, orderId.length());


        // TODO 标准化

        // TODO 其他

        // 如果某个订单数据结束
        if ("end".equals(driverId)) {
            builder
                    .driverId(afterJsonObj.getString("driver_id"))
                    .orderId(afterJsonObj.getString("order_id"))
                    .orderIdDesensitization(orderIdDesensitization);
        } else {
            builder
                    .driverId(afterJsonObj.getString("driver_id"))
                    .orderId(afterJsonObj.getString("order_id"))
                    .orderIdDesensitization(orderIdDesensitization)
                    .time(afterJsonObj.getString("time"))
                    .lng(afterJsonObj.getString("lng"))
                    .lat(afterJsonObj.getString("lat"));
        }
        out.collect(builder.build());
    }
}
